package com.hhf.rpc.core.client.impl;

import cn.hutool.core.util.IdUtil;
import com.hhf.rpc.core.RpcApplication;
import com.hhf.rpc.core.constant.ProtocolConstant;
import com.hhf.rpc.core.enums.ProtocolMessageSerializerEnum;
import com.hhf.rpc.core.enums.ProtocolMessageTypeEnum;
import com.hhf.rpc.core.handler.TcpBufferHandlerWrapper;
import com.hhf.rpc.core.model.RpcRequest;
import com.hhf.rpc.core.model.RpcResponse;
import com.hhf.rpc.core.model.ServiceMetaInfo;
import com.hhf.rpc.core.protocol.ProtocolMessage;
import com.hhf.rpc.core.protocol.ProtocolMessageDecoder;
import com.hhf.rpc.core.protocol.ProtocolMessageEncoder;
import io.vertx.core.Vertx;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.net.NetClient;
import io.vertx.core.net.NetSocket;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

/**
 * @Author hhf
 * @Date 2024/10/19
 * @Description
 */
@Slf4j
public class VertxTcpClient {

    public static RpcResponse doRequest(ServiceMetaInfo serviceMetaInfo, RpcRequest rpcRequest) throws InterruptedException, ExecutionException {

        String host = serviceMetaInfo.getServiceHost();
        Integer port = serviceMetaInfo.getServicePort();

        CompletableFuture<RpcResponse> responseFuture = new CompletableFuture<>();

        //1.创建vertx以tpc传输的客户端
        Vertx vertx = Vertx.vertx();
        NetClient client = vertx.createNetClient();
        //2.连接指定的服务端
        log.info("连接服务器:{}:{}", host, port);
        client.connect(port, host, result -> {
            if (result.succeeded()) {
                log.info("connect to TCP server");
                NetSocket socket = result.result();
                //构造消息
                ProtocolMessage<RpcRequest> protocolMessage = new ProtocolMessage<>();
                ProtocolMessage.Header header = new ProtocolMessage.Header();
                header.setMagic(ProtocolConstant.PROTOCOL_MAGIC);
                header.setVersion(ProtocolConstant.PROTOCOL_VERSION);
                header.setType((byte) ProtocolMessageTypeEnum.REQUEST.getKey());
                header.setSerializer((byte) ProtocolMessageSerializerEnum.getSerializerByValue(RpcApplication.getRpcConfig().getSerializerKey()).getKey());
                header.setRequestId(IdUtil.getSnowflakeNextId());

                protocolMessage.setHeader(header);
                protocolMessage.setBody(rpcRequest);
                //编码
                try {
                    Buffer requestBuffer = ProtocolMessageEncoder.encode(protocolMessage);
                    //发送请求
                    socket.write(requestBuffer);
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
                //装饰器模式,加强=>解决粘包半包问题
                TcpBufferHandlerWrapper bufferHandlerWrapper = new TcpBufferHandlerWrapper(buffer -> {
                    log.info("client receive response");
                    try {
                        ProtocolMessage<RpcResponse> responseProtocolMessage = (ProtocolMessage<RpcResponse>) ProtocolMessageDecoder.decode(buffer);
                        responseFuture.complete(responseProtocolMessage.getBody());
                    } catch (IOException e) {
                        throw new RuntimeException("协议消息解码错误");
                    }
                });

                //接受响应
                socket.handler(bufferHandlerWrapper);
            } else {
                log.info("fail connect to TCP server:", result.cause());
            }
        });

        RpcResponse rpcResponse = responseFuture.get();
        //关闭连接
        client.close();
        return rpcResponse;

    }
}
